package com.kun.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.junit.jupiter.api.Test;

import java.util.Collection;
import java.util.List;

/**
 * @author FangKun
 * @create 2024-01-29-13:43
 */
public class SimpleRocketMQ {

    @Test
    public void producer() throws Exception {

        DefaultMQProducer producer = new DefaultMQProducer("ooxx");
        producer.setNamesrvAddr("192.168.81.21:9876");
        producer.start();

        for (int i = 0; i < 10; i++) {
            Message message = new Message();
            message.setTopic("wula");
            message.setTags("tagA");
            message.setBody(("ooxx"+i).getBytes());
            message.setWaitStoreMsgOK(true);

            //发送消息
            /*
            SendResult result = producer.send(message);
            System.out.println(result);

            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(sendResult);
                }

                @Override
                public void onException(Throwable throwable) {

                }
            });

            producer.sendOneway(message);
            */

            MessageQueue mq = new MessageQueue("wula","node1",0);
            SendResult result = producer.send(message, mq);
            System.out.println(result);

        }

    }

    @Test
    public void consumerPull() throws Exception {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("xxxx");
        consumer.setNamesrvAddr("192.168.81.22:9876");
        consumer.start();

        Collection<MessageQueue> messageQueues = consumer.fetchMessageQueues("wula");
        System.out.println("=============::queues::===========");
        messageQueues.forEach(messageQueue -> System.out.println(messageQueue));

        System.out.println("=============::poll::===========");
        consumer.assign(messageQueues);
        List<MessageExt> poll = consumer.poll();
        poll.forEach(s -> System.out.println(s));
    }

    @Test
    public void consumerPush() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_ox");
        consumer.setNamesrvAddr("192.168.81.21:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe("wula","*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //拿过来的是一个小集合
                list.forEach( messageExt -> {
                    System.out.println(messageExt);
                    byte[] body = messageExt.getBody();
                    String msg = new String(body);
                    System.out.println(msg);
                });
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.in.read();
    }

    @Test
    public void admin() throws Exception {

        DefaultMQAdminExt admin = new DefaultMQAdminExt();
        admin.setNamesrvAddr("192.168.81.22:9876");
        admin.start();

        TopicList topicList = admin.fetchAllTopicList();
        for (String s : topicList.getTopicList()) {
            System.out.println(s);
        }

        System.out.println("=========topic route | info===========");

        TopicRouteData wula = admin.examineTopicRouteInfo("wula");
        System.out.println(wula);

    }

}
